package net.bwie.flink;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * FlinkSQL对app行为日志数据进行处理分析
 * @author xuanyu
 * @date 2025/10/27
 */
public class _06SqlAppLog {

	public static void main(String[] args) {
		// 1.表执行环境-tabEnv
		EnvironmentSettings settings = EnvironmentSettings
			.newInstance()
			.inStreamingMode()
			.build();
		TableEnvironment tabEnv = TableEnvironment.create(settings);

		// 2. 输入表-inputTable
		tabEnv.executeSql(
			"CREATE TABLE app_log_kafka_source (\n" +
				"    `common` MAP<STRING, STRING>,\n" +
				"    `page` MAP<STRING, STRING>,\n" +
				"    `start` MAP<STRING, STRING>,\n" +
				"    `err` MAP<STRING, STRING>,\n" +
				"    `displays` ARRAY<MAP<STRING, STRING>>,\n" +
				"    `actions` ARRAY<MAP<STRING, STRING>>,\n" +
				"    `ts` BIGINT\n" +
				") WITH (\n" +
				"  'connector' = 'kafka',\n" +
				"  'topic' = 'topic-log',\n" +
				"  'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
				"  'properties.group.id' = 'sql-app-g1',\n" +
				"  'scan.startup.mode' = 'earliest-offset',\n" +
				"  'format' = 'json',\n" +
				"  'json.fail-on-missing-field' = 'false',\n" +
				"  'json.ignore-parse-errors' = 'true'\n" +
				")"
		);
		// tabEnv.sqlQuery("SELECT * FROM app_log_kafka_source").execute().print();

		// 3.查询语句-select
		// todo 3-1.实时统计PV浏览量
		Table pvTable = tabEnv.sqlQuery(
			"SELECT\n" +
				"    'page_view' AS report_type,\n" +
				"    count(1) AS report_value\n" +
				"FROM app_log_kafka_source\n" +
				"WHERE `page` IS NOT NULL\n" +
				"    AND page['page_id'] IS NOT NULL"
		);
		// pvTable.execute().print();

		// todo 3-2.实时统计各个页面PV浏览量
		Table pagePvTable = tabEnv.sqlQuery(
			"SELECT\n" +
				"    page['page_id'] AS report_type,\n" +
				"    count(1) AS report_value\n" +
				"FROM app_log_kafka_source\n" +
				"WHERE `page` IS NOT NULL\n" +
				"  AND page['page_id'] IS NOT NULL\n" +
				"GROUP BY page['page_id']"
		);
		// pagePvTable.execute().print();

		// 4.输出表-outputTable
		tabEnv.executeSql(
			"CREATE TABLE app_pv_mysql_sink (\n" +
				"    report_type STRING,\n" +
				"    report_value BIGINT,\n" +
				"    PRIMARY KEY (report_type) NOT ENFORCED\n" +
				") WITH (\n" +
				"   'connector' = 'jdbc',\n" +
				"   'url' = 'jdbc:mysql://node101:3306/db_flink',\n" +
				"   'driver' = 'com.mysql.jdbc.Driver',\n" +
				"   'table-name' = 'app_pv_report',\n" +
				"   'username' = 'root',\n" +
				"   'password' = '123456'\n" +
				")"
		);

		// 5.插入语句-insert
		tabEnv.createTemporaryView("pv_table", pvTable);
		tabEnv.executeSql(
			"INSERT INTO app_pv_mysql_sink\n" +
				"SELECT report_type, report_value  FROM pv_table"
		);

		tabEnv.createTemporaryView("page_pv_table", pagePvTable);
		tabEnv.executeSql(
			"INSERT INTO app_pv_mysql_sink\n" +
				"SELECT report_type, report_value  FROM page_pv_table"
		);

	}

}
